package com.spark.gaea.config.service.easyzk;

import com.google.common.base.*;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundPathable;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.utils.ZKPaths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PreDestroy;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class ConfigNode {
	static final Logger LOG = LoggerFactory.getLogger(ConfigNode.class);
	private ConfigProfile configProfile;
	private String node;
	private KeyLoadingMode keyLoadingMode;
	private Set<String> keysSpecified;
	private CuratorFramework client;
	private ConfigLocalCache configLocalCache;
	private final Map<String, String> properties;

	public void setConfigLocalCache(final ConfigLocalCache configLocalCache) {
		this.configLocalCache = configLocalCache;
	}

	protected ConfigNode(final ConfigProfile configProfile, final String node) {
		this.properties = Maps.newConcurrentMap();
		this.configProfile = configProfile;
		this.node = node;
	}

	public void defineKeyLoadingPattern(final KeyLoadingMode keyLoadingMode, final Set<String> keysSpecified) {
		this.keyLoadingMode = (KeyLoadingMode) Preconditions.checkNotNull((Object) keyLoadingMode);
		this.keysSpecified = (Set<String>) ((keysSpecified != null) ? Sets.newHashSet((Iterable) keysSpecified) : null);
	}

	protected void initConfigNode() {
		this.client = CuratorFrameworkFactory
				.newClient(this.configProfile.getConnectStr(), this.configProfile.getRetryPolicy());
		this.client.getCuratorListenable().addListener(new ConfigNodeEventListener(this));
		this.client.start();
		this.loadNode();
		if (this.configLocalCache != null) {
			this.configLocalCache.saveLocalCache(this, this.node);
		}
	}

	void loadNode() {
		final String nodePath = ZKPaths.makePath(this.configProfile.getRootNode(), this.node);
		if (ConfigNode.LOG.isDebugEnabled()) {
			ConfigNode.LOG.debug("Loading properties for node: [{}], with loading mode: [{}] and keys specified: [{}]",
					nodePath, this.keyLoadingMode, this.keysSpecified);
		}
		final GetChildrenBuilder childrenBuilder = this.client.getChildren();
		try {
			this.properties.clear();
			final List<String> children = (List<String>) ((BackgroundPathable) childrenBuilder.watched())
					.forPath(nodePath);
			if (children != null) {
				for (final String child : children) {
					this.loadKey(ZKPaths.makePath(nodePath, child));
				}
			}
		} catch (Exception e) {
			throw Throwables.propagate(e);
		}
	}

	void loadKey(final String nodePath) throws Exception {
		final String nodeName = ZKPaths.getNodeFromPath(nodePath);
		switch (this.keyLoadingMode) {
			case INCLUDE: {
				if (!this.keysSpecified.contains(nodeName)) {
					return;
				}
				break;
			}
			case EXCLUDE: {
				if (this.keysSpecified.contains(nodeName)) {
					return;
				}
				break;
			}
		}
		final GetDataBuilder data = this.client.getData();
		final String childValue = new String((byte[]) ((BackgroundPathable) data.watched()).forPath(nodePath),
				Charsets.UTF_8);
		if (ConfigNode.LOG.isDebugEnabled()) {
			ConfigNode.LOG.debug("Loading data: key[{}] - value[{}]", nodeName, childValue);
		}
		this.properties.put(nodeName, childValue);
	}

	@PreDestroy
	private void destroy() {
		if (this.client != null) {
			this.client.close();
		}
	}

	public String getProperty(final String key) {
		return this.properties.get(key);
	}

	public String getProperty(final String key, final String defaultValue) {
		final String value = this.getProperty(key);
		if (Strings.isNullOrEmpty(value)) {
			return defaultValue;
		}
		return value;
	}

	public String getNode() {
		return this.node;
	}

	public ConfigLocalCache getConfigLocalCache() {
		return this.configLocalCache;
	}

	public Map<String, String> exportProperties() {
		return (Map<String, String>) Maps.newHashMap((Map) this.properties);
	}

	public void importProperties(final Map<String, String> imports) {
		this.properties.putAll(imports);
	}

	@Override
	public String toString() {
		return MoreObjects.toStringHelper(this).add("configProfile", this.configProfile).add("node", this.node)
				.add("keyLoadingMode", this.keyLoadingMode).add("keysSpecified", this.keysSpecified)
				.add("properties", this.properties).toString();
	}

	public enum KeyLoadingMode {
		NONE, INCLUDE, EXCLUDE
	}
}
